#include "url.h"
#include "pending_schema.h"
#include "schema_operate.h"
#include "download_manager.h"
#include <assert.h>
#include "log.h"

#include "conf.h"

//#define MYSQL_PENDING_SCHEMA_SERVER 	"localhost"
//#define MYSQL_PENDING_SCHEMA_USER	 	"webroot"
//#define MYSQL_PENDING_SCHEMA_PASSWORD 	"123456"
//#define MYSQL_PENDING_SCHEMA_DB 		"pending_schema"
#define SELECT_PENDING_TO_DOWNLOAD_COUNT 20

pending_schema_t g_pending_schema;

void quit_pthread_pending_schema(MYSQL* mysql, 
	pthread_mutex_t& locker, pthread_cond_t& cond )
{
	pthread_mutex_destroy(&locker);
	pthread_cond_destroy(&cond);
	mysql_close(mysql);
}

void* pthread_pending_schema(void* data)
{
	pthread_detach(pthread_self());
	
	int i = (int)data;
	list<table_op_t*>& table_ops = g_pending_schema.table_ops[i];
	list<table_op_t*>::iterator it;
	pthread_mutex_t& locker = g_pending_schema.lockers[i];
	pthread_cond_t&  cond = g_pending_schema.conds[i];
	
	MYSQL* mysql = mysql_init(NULL);
	if(!mysql_real_connect(mysql, 
		g_conf.pending.host.c_str(),
		g_conf.pending.user.c_str(),
		g_conf.pending.password.c_str(),
		g_conf.pending.name.c_str(), 0, NULL, 0))
	{
		printf("error, mysql connect pending_schema failed\n");
		return (void*)-1;
	}
	while(1)
	{
		pthread_mutex_lock(&locker);
		while(table_ops.size()==0)
		{
			pthread_cond_wait(&cond, &locker);
		}
		table_op_t* table_op = table_ops.front();
		table_ops.pop_front();
		pthread_mutex_unlock(&locker);
		
		if( table_op == 0)
			continue;
		
		eTABLE_OP op = table_op->op;
		schema_t* schema = (schema_t*)table_op->schema;
		url_t* url = (url_t*)schema->data;
		assert(url != NULL);
		// insert url to db ..
		int ret = 0;
		if( op == INSERT_TABLE)
		{
			ret = insert_to_pending(mysql, url);	
		}
		else if( op == COUNT_TABLE )
		{
			ret = exist_in_pending(mysql, url) ;// return 0, will go on
		}
		else if( op == DELETE_TABLE)
		{
			ret = delete_from_pending(mysql, url);
		}
		else
		{
			assert(false);
		}
		
		go_next_schema(schema, ret);
		
		del_table_op(table_op);
		if(g_pending_schema.quit)
		{
			pthread_mutex_lock(&locker);
			if(table_ops.size()==0)
			{
				pthread_mutex_unlock(&locker);
				break;
			}
			pthread_mutex_unlock(&locker);
		}
	}	
	quit_pthread_pending_schema(mysql,locker, cond);
	return (void*)0;
}

void quit_pthread_select_pending_to_download(MYSQL* mysql, list<url_t*>& urls)
{
	if(mysql == NULL)
		return;
		
	list<url_t*>::iterator it = urls.begin();
	for(; it != urls.end(); it++)
	{
		release_url((void*)(*it));
	}
	urls.clear();
	mysql_close(mysql);
}
void* pthread_select_pending_to_download(void*)
{
	pthread_detach(pthread_self());
	
	list<url_t*> urls;
	
	MYSQL* mysql = mysql_init(NULL);
	if(!mysql_real_connect(mysql, 
		g_conf.pending.host.c_str(),
		g_conf.pending.user.c_str(),
		g_conf.pending.password.c_str(),
		g_conf.pending.name.c_str(), 0, NULL, 0))
	{
		printf("error, mysql connect failed\n");
		return (void*)-1;
	}
	while(1)
	{
		sleep(1);//
		
		select_from_pending(mysql, SELECT_PENDING_TO_DOWNLOAD_COUNT, urls);
		add_urls_to_download_manager(urls, &g_download_manager);
		urls.clear(); //clear directly, the new url is in g_download_manager
		if(g_pending_schema.quit)
		{
			break;
		}
	}	
	quit_pthread_select_pending_to_download(mysql,urls);
	return (void*)0;
}

int init_pending_schema(pending_schema_t& pending_schema)
{
	
	pending_schema.quit = false;
	for(int i = 0; i < PENDING_SCHEMA_THREAD_COUNT; i++)
	{
		pthread_mutex_init(&pending_schema.lockers[i], NULL);
	}
	for(int i = 0; i < PENDING_SCHEMA_THREAD_COUNT; i++)
	{
		pthread_cond_init(&pending_schema.conds[i], NULL);
	}
	pthread_t th_pending_schema[PENDING_SCHEMA_THREAD_COUNT];
	for(int i = 0 ; i < PENDING_SCHEMA_THREAD_COUNT; i++)
	{
		pthread_create(&th_pending_schema[i], NULL, &pthread_pending_schema,
					   (void*)i);
	}
	pthread_t th_pthread_select_pending_to_download;
	pthread_create(&th_pthread_select_pending_to_download, NULL,
			&pthread_select_pending_to_download, 0);
	return 0;
}

int op_pending_schema(schema_t* schema,  eTABLE_OP op)
{
	if( schema == 0 )
		assert(false);
	//approximate the min	
	int index = rand()%PENDING_SCHEMA_THREAD_COUNT;
	if(index >= PENDING_SCHEMA_THREAD_COUNT)
	{
		index = PENDING_SCHEMA_THREAD_COUNT -1;
	}

	table_op_t* table_op = creat_table_op(schema, op);
	pthread_mutex_lock(&g_pending_schema.lockers[index]);
	g_pending_schema.table_ops[index].push_back(table_op);
	pthread_cond_signal(&g_pending_schema.conds[index]);
	pthread_mutex_unlock(&g_pending_schema.lockers[index]);	
	return 0;
}

int add_to_pending_schema(void* data)
{
	assert(data != NULL);
	return op_pending_schema((schema_t*)data, INSERT_TABLE);
}

int remove_from_pending_schema(void* data)
{
	assert(data != NULL);
	return op_pending_schema((schema_t*)data, DELETE_TABLE);
}

void quit_pending_schema(pending_schema_t* pending_schema)
{
	pending_schema->quit = true;
}
int exist_in_pending_schema(void* data)
{
	assert(data != NULL);
	return op_pending_schema((schema_t*)data,COUNT_TABLE);
}



